/*
 * Copyright (c) 2003-2010, Live Media Pty. Ltd.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without modification, are permitted
 * provided that the following conditions are met:
 *
 *  1. Redistributions of source code must retain the above copyright notice, this list of
 *     conditions and the following disclaimer.
 *  2. Redistributions in binary form must reproduce the above copyright notice, this list of
 *     conditions and the following disclaimer in the documentation and/or other materials
 *     provided with the distribution.
 *  3. Neither the name of Live Media nor the names of its contributors may be used to endorse
 *     or promote products derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
 * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package com.colony.concurrent;

/**
 *  A Queue class which extends the simple Linked List class.  Adds functionality
 *  of blocking reads (producer/consumer) method.  Where multiple threads are blocked
 *  on read a round robin approach is created.  If no threads are blocked and event
 *  semaphores have been set, an event semaphore will be posted.  If the semaphore 
 *  does not have a listener the message will be lost.
 */

public class Queue
extends List
{
   private final int DEFAULT_SIZE = 50;

   boolean    _notify;
   EventSem[] _notifySems;
   int        _notifyCount;
   int        _notifyRobin;

   int      _waiting;
   EventSem _waitingSem;

   public Queue()
   {
      _notify = false;
      _notifySems = new EventSem[ DEFAULT_SIZE ];
      _notifyCount = 0;
      _notifyRobin = 0;

      _waiting = 0;
      _waitingSem = new EventSem();
   }

   /**
    * write an object to the queue and wake up a thread is any are blocked on read.
    */

   public void write( Object m )
   {
      synchronized ( this )
      {
          add( m );
          
          if ( _waiting > 0 ) 
          {
             _waitingSem.post();
             return;
          }

          if ( _notify )
          {
             int currentSem = _notifyRobin;
	     while ( _notifySems[ _notifyRobin ].query() > 0 ) 
             {
                _notifyRobin++;
                if (_notifyRobin >= _notifyCount ) _notifyRobin = 0;

                // This is dangerous if a problem occurs and we think all the semaphores have
                // been posted.  We are throwing the message on the queue and forgetting about it.
                if (_notifyRobin == currentSem ) return;  
             }

             _notifySems[ _notifyRobin ].post();
             //_notifyRobin++;
             return;
          }
      }
   }

   /**
    *  read an item from the Queue.  Block can be set to true to make the reader
    *  block if no messages are available.
    */

   public Object read( boolean block )
   {
      synchronized ( this )
      {
         if ( isEmpty() )
         {
            if ( !block )
               return null;

            _waiting++;
            while ( isEmpty() ) _waitingSem.waitSem();

            _waitingSem.reset();
            _waiting--;
         }
         return remove();
      }
   }

   /**
    *  set a semaphore to be posted when a message is placed on the queue.
    */

   public void setEventSem( EventSem sem )
   {
      synchronized( this )
      {
          _notifySems[ _notifyCount ] = sem;
          _notifyCount++;
          _notify = true;
      }
   }

}